package cn.itcast.dmp.runner

import cn.itcast.dmp.area.AreaProcessor
import cn.itcast.dmp.config.AppConfigHelper
import cn.itcast.dmp.untils.SparkSessionUtils
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 根据经纬度, 查询高德的 API 获取商圈信息, 生成数据库放置商圈信息
  */
object BusinessAreasRunner extends Logging{

	// ETL后ODS表：广告数据
	private val SOURCE_TABLE_NAME = AppConfigHelper.AD_MAIN_TABLE_NAME
	// 商圈信息库表
	private val SINK_TABLE_NAME = AppConfigHelper.BUSINESS_AREAS_TABLE_NAME

	def main(args: Array[String]): Unit = {

		// 第一步、构建SparkSession实例对象
		val spark: SparkSession = SparkSessionUtils.createSparkSession(this.getClass)

		// 第二步、读取ODS表的数据
		import cn.itcast.dmp.untils.KuduUtils._
		val odsDF: DataFrame = spark.readKuduTable(SOURCE_TABLE_NAME) match {
			case Some(df) => df
			case None => println("ERROR: ODS表无数据,结束执行"); return
		}

		// 第三步、构建商圈库
		val areaDF: DataFrame = AreaProcessor.processData(odsDF)

		// 第四步、保存商圈库信息到Kudu表中
		spark.createKuduTable(SINK_TABLE_NAME, areaDF.schema, Seq("geo_hash"), isDelete = false)
		areaDF.saveAsKuduTable(SINK_TABLE_NAME)

		// 第五步、应用结束，关闭资源
		spark.stop()
	}

}
